import struct
import time

try:
    from TestToolApi import Debug
except ModuleNotFoundError:
    from .TestToolApi import Debug
from multiprocessing import Process, Queue as ProcQueue, Event
from threading import Thread
from queue import Queue as ThreadQueue
from multiprocessing.sharedctypes import Value
import select
import asyncio
import traceback
import socket


class Video_StoC_Main(Process):

    def __init__(
            self,
            recv_port_main: int,
            sent_port_main: int,
            sent_address: str = '169.254.171.1',
            logging=True
    ):
        super(Video_StoC_Main, self).__init__()
        if logging:
            self.__glDebugQueue = ProcQueue(10)
        else:
            self.__glDebugQueue = None
        self.__recvPortMain = recv_port_main
        self.__sentPortMain = sent_port_main
        self.__sentAddress = sent_address

        self.__mainThreadDebug = Debug(self.__glDebugQueue, True, "视频流通道进程(主)|(STC)")
        self.__ErrorStatus = Event()
        self.__nowMainNums = Value('Q', 0, lock=False)
        self.__mainMs = Value('f', 0.0, lock=False)

    async def __taskServerMainRecv(self, recv_port: int, channel: asyncio.Queue):
        self.__mainThreadDebug.pushInfo("[视频流接收器] 视频流接收协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        server.bind(('10.1.0.5', recv_port))
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLIN)
        self.__mainThreadDebug.pushInfo("[视频流接收器] 已进入接收循环")
        lastTime = time.time()
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    await asyncio.sleep(0)
                else:
                    recvBytes, _ = server.recvfrom(65535)
                    nowTime = time.time()
                    self.__mainMs.value = round((nowTime - lastTime) * 1000, 2)
                    self.__nowMainNums.value += 1
                    lastTime = nowTime
                    await channel.put(recvBytes)
            except Exception as e:
                self.__mainThreadDebug.pushInfo(
                    f"[视频流接收器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainThreadDebug.pushInfo("[视频流接收器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainThreadDebug.pushInfo("[视频流接收器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    async def __taskClientMainSent(self, sent_port: int, sent_address: str, channel: asyncio.Queue):
        self.__mainThreadDebug.pushInfo("[视频流发送器] 视频流发送协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLOUT)
        self.__mainThreadDebug.pushInfo("[视频流发送器] 已进入发送循环...")
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    await asyncio.sleep(0)
                else:
                    if channel.empty():
                        await asyncio.sleep(0)
                    else:
                        chunk = await channel.get()
                        server.sendto(chunk, (sent_address, sent_port))
            except Exception as e:
                self.__mainThreadDebug.pushInfo(
                    f"[视频流发送器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainThreadDebug.pushInfo("[视频流发送器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainThreadDebug.pushInfo("[视频流发送器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    def __mainThread(
            self,
            recv_port_main,
            sent_port_main,
            sent_address
    ):
        self.__mainThreadDebug.pushInfo("正在初始化主线程...", 'Debug')
        myself = self
        loop = asyncio.new_event_loop()
        mainVideoQueue = asyncio.Queue(20, loop=loop)

        taskServerMainRecv = loop.create_task(self.__taskServerMainRecv(recv_port_main, mainVideoQueue))
        taskClientMainSent = loop.create_task(self.__taskClientMainSent(sent_port_main, sent_address, mainVideoQueue))

        async def main(__self=myself):
            await asyncio.gather(
                taskServerMainRecv, taskClientMainSent
            )

        self.__mainThreadDebug.pushInfo("主线程开始运行", 'Debug')
        loop.run_until_complete(main())
        self.__ErrorStatus.set()
        self.__mainThreadDebug.pushInfo("退出该线程!", "Debug")

    def run(self) -> None:
        dataSentThread = Thread(
            target=self.__mainThread,
            args=(self.__recvPortMain, self.__sentPortMain, self.__sentAddress)
        )
        dataSentThread.setDaemon(True)
        dataSentThread.start()
        dataSentThread.join()

    def openServer(self):
        self.start()

    def closeServer(self):
        self.__ErrorStatus.set()

    def isOpened(self):
        return self.is_alive()

    def getNums(self):
        return self.__nowMainNums.value

    def getMs(self):
        return self.__mainMs.value


class Video_StoC_Sub(Process):

    def __init__(
            self,
            recv_port_sub: int,
            sent_port_sub: int,
            sent_address: str = '169.254.171.1',
            logging=True
    ):
        super(Video_StoC_Sub, self).__init__()
        if logging:
            self.__glDebugQueue = ProcQueue(10)
        else:
            self.__glDebugQueue = None
        self.__recvPortSub = recv_port_sub
        self.__sentPortSub = sent_port_sub
        self.__sentAddress = sent_address

        self.__mainThreadDebug = Debug(self.__glDebugQueue, True, "视频流通道进程(副)|(STC)")
        self.__ErrorStatus = Event()
        self.__nowSubNums = Value('Q', 0, lock=False)
        self.__subMs = Value('f', 0.0, lock=False)

    async def __taskServerSubRecv(self, recv_port: int, channel: asyncio.Queue):
        self.__mainThreadDebug.pushInfo("[视频流接收器] 视频流接收协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        server.bind(('10.1.0.5', recv_port))
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLIN)
        self.__mainThreadDebug.pushInfo("[视频流接收器] 已进入接收循环")
        lastTime = time.time()
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    await asyncio.sleep(0)
                else:
                    recvBytes, _ = server.recvfrom(65535)
                    nowTime = time.time()
                    self.__subMs.value = round((nowTime - lastTime) * 1000, 2)
                    self.__nowSubNums.value = 1
                    lastTime = nowTime
                    await channel.put(recvBytes)
            except Exception as e:
                self.__mainThreadDebug.pushInfo(
                    f"[视频流接收器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainThreadDebug.pushInfo("[视频流接收器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainThreadDebug.pushInfo("[视频流接收器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    async def __taskClientSubSent(self, sent_port: int, sent_address: str, channel: asyncio.Queue):
        self.__mainThreadDebug.pushInfo("[视频流发送器] 视频流发送协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLOUT)
        self.__mainThreadDebug.pushInfo("[视频流发送器] 已进入发送循环...")
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    await asyncio.sleep(0)
                else:
                    chunk = await channel.get()
                    server.sendto(chunk, (sent_address, sent_port))
                    await asyncio.sleep(0)
            except Exception as e:
                self.__mainThreadDebug.pushInfo(
                    f"[视频流发送器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainThreadDebug.pushInfo("[视频流发送器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainThreadDebug.pushInfo("[视频流发送器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    def __mainThread(
            self,
            recv_port_sub,
            sent_port_sub,
            sent_address
    ):
        self.__mainThreadDebug.pushInfo("正在初始化主线程...", 'Debug')
        myself = self
        loop = asyncio.new_event_loop()
        subVideoQueue = asyncio.Queue(20, loop=loop)

        taskServerSubRecv = loop.create_task(self.__taskServerSubRecv(recv_port_sub, subVideoQueue))
        taskClientSubSent = loop.create_task(self.__taskClientSubSent(sent_port_sub, sent_address, subVideoQueue))

        async def main(__self=myself):
            await asyncio.gather(
                taskServerSubRecv, taskClientSubSent
            )

        self.__mainThreadDebug.pushInfo("主线程开始运行", 'Debug')
        loop.run_until_complete(main())
        self.__ErrorStatus.set()
        self.__mainThreadDebug.pushInfo("总进程遇到错误, 退出该线程!", "Warning")

    def run(self) -> None:
        dataSentThread = Thread(
            target=self.__mainThread,
            args=(self.__recvPortSub, self.__sentPortSub, self.__sentAddress)
        )
        dataSentThread.setDaemon(True)
        dataSentThread.start()
        dataSentThread.join()

    def openServer(self):
        self.start()

    def closeServer(self):
        self.__ErrorStatus.set()

    def isOpened(self):
        return self.is_alive()

    def getNums(self):
        return self.__nowSubNums.value

    def getMs(self):
        return self.__subMs.value


class Video_CtoS_Channel(Process):

    def __init__(
            self,
            sent_port: int,
            recv_port: int,
            sent_address: str = '10.1.0.4',
            logging=True
    ):
        super(Video_CtoS_Channel, self).__init__()
        if logging:
            self.__glDebugQueue = ProcQueue(10)
        else:
            self.__glDebugQueue = None
        self.__sentPort = sent_port
        self.__sentAddress = sent_address
        self.__recvPort = recv_port
        self.__mainThreadDebug = Debug(self.__glDebugQueue, True, "视频流通道进程(CTS)")
        self.__ErrorStatus = Event()

    async def __taskVideoRecv(self, recv_port: int, channel: asyncio.Queue):
        self.__mainThreadDebug.pushInfo("[视频流接收器] 客户端视频流接收协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        server.bind(('10.1.0.5', recv_port))
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLIN)
        self.__mainThreadDebug.pushInfo("[视频流接收器] 已进入接收循环")
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    await asyncio.sleep(0)
                else:
                    recvBytes, _ = server.recvfrom(65535)
                    await channel.put(recvBytes)
            except Exception as e:
                self.__mainThreadDebug.pushInfo(
                    f"[视频流接收器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainThreadDebug.pushInfo("[视频流接收器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainThreadDebug.pushInfo("[视频流接收器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    async def __taskVideoSent(self, sent_port: int, sent_address: str, channel: asyncio.Queue):
        self.__mainThreadDebug.pushInfo("[视频流发送器] 客户端视频流发送协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLOUT)
        self.__mainThreadDebug.pushInfo("[视频流发送器] 已进入发送循环...")
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    await asyncio.sleep(0)
                else:
                    chunk = await channel.get()
                    server.sendto(chunk, (sent_address, sent_port))
            except Exception as e:
                self.__mainThreadDebug.pushInfo(
                    f"[视频流发送器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainThreadDebug.pushInfo("[视频流发送器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainThreadDebug.pushInfo("[视频流发送器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    def __mainThread(self, recv_port, sent_port, sent_address):
        self.__mainThreadDebug.pushInfo("正在初始化主线程...", 'Debug')
        myself = self
        loop = asyncio.new_event_loop()
        videoQueue = asyncio.Queue(20, loop=loop)

        taskVideoRecv = loop.create_task(self.__taskVideoRecv(recv_port, videoQueue))
        taskVideoSent = loop.create_task(self.__taskVideoSent(sent_port, sent_address, videoQueue))

        async def main(__self=myself):
            await asyncio.gather(
                taskVideoRecv, taskVideoSent
            )

        self.__mainThreadDebug.pushInfo("主线程开始运行", 'Debug')
        loop.run_until_complete(main())
        self.__ErrorStatus.set()
        self.__mainThreadDebug.pushInfo("总进程遇到错误, 退出该线程!", "Warning")

    def run(self) -> None:
        dataSentThread = Thread(
            target=self.__mainThread,
            args=(self.__recvPort, self.__sentPort, self.__sentAddress)
        )
        dataSentThread.setDaemon(True)
        dataSentThread.start()
        dataSentThread.join()

    def openServer(self):
        self.start()

    def closeServer(self):
        self.__ErrorStatus.set()

    def isOpened(self):
        return self.is_alive()


class Audio_STC_Channel(Process):

    def __init__(
            self,
            recv_port_s: int,
            sent_port_c: int,
            sent_address_c: str = "169.254.171.1",
            logging=True
    ):
        super(Audio_STC_Channel, self).__init__()
        if logging:
            self.__glDebugQueue = ProcQueue(10)
        else:
            self.__glDebugQueue = None
        self.__sentPortC = sent_port_c
        self.__sentAddressC = sent_address_c
        self.__recvPortS = recv_port_s
        self.__mainThreadDebug = Debug(self.__glDebugQueue, True, "音频流通道进程(STC)")
        self.__ErrorStatus = Event()
        self.__nowNums = Value('Q', 0, lock=True)
        self.__nowMs = Value("f", 0.0, lock=True)

    def __taskAudioServerRecv(self, recv_port: int, channel: ThreadQueue):
        self.__mainThreadDebug.pushInfo("[音频流接收器] 服务端音频流接收协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        server.bind(('10.1.0.5', recv_port))
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLIN)
        self.__mainThreadDebug.pushInfo("[音频流接收器] 已进入接收循环")
        lastTime = time.time()
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    time.sleep(0)
                else:
                    recvBytes, _ = server.recvfrom(65535)
                    nowTime = time.time()
                    self.__nowMs.value = round((nowTime - lastTime) * 1000, 2)
                    self.__nowNums.value += 1
                    lastTime = nowTime
                    channel.put(recvBytes)
            except Exception as e:
                self.__mainThreadDebug.pushInfo(
                    f"[音频流接收器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainThreadDebug.pushInfo("[音频流接收器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainThreadDebug.pushInfo("[音频流接收器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    def __taskAudioClientSent(self, sent_port: int, sent_address: str, channel: ThreadQueue):
        self.__mainThreadDebug.pushInfo("[音频流发送器] 客户端音频流发送协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLOUT)
        self.__mainThreadDebug.pushInfo("[音频流发送器] 已进入发送循环...")
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    time.sleep(0)
                else:
                    chunk = channel.get()
                    server.sendto(chunk, (sent_address, sent_port))
            except Exception as e:
                self.__mainThreadDebug.pushInfo(
                    f"[音频流发送器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainThreadDebug.pushInfo("[音频流发送器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainThreadDebug.pushInfo("[音频流发送器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    def run(self) -> None:
        self.__mainThreadDebug.pushInfo("正在初始化主进程...", 'Debug')
        serverAudioChannel = ThreadQueue(20)
        audioServerRecv = Thread(
            target=self.__taskAudioServerRecv,
            args=(self.__recvPortS, serverAudioChannel)
        )
        audioClientSent = Thread(
            target=self.__taskAudioClientSent,
            args=(self.__sentPortC, self.__sentAddressC, serverAudioChannel)
        )
        audioServerRecv.setDaemon(True)
        audioClientSent.setDaemon(True)
        audioServerRecv.start()
        audioClientSent.start()
        self.__mainThreadDebug.pushInfo("线程运行中...", 'Debug')
        audioClientSent.join()
        audioServerRecv.join()
        self.__mainThreadDebug.pushInfo("主进程运行结束!", "Warning")

    def openServer(self):
        self.start()

    def closeServer(self):
        self.__ErrorStatus.set()

    def isOpened(self):
        return self.is_alive()

    def getNums(self):
        return self.__nowNums.value

    def getMs(self):
        return self.__nowMs.value


class Audio_CTS_Channel(Process):

    def __init__(
            self,
            sent_port_s: int,
            recv_port_c: int,
            sent_address_s: str = "10.1.0.4",
            logging=True
    ):
        super(Audio_CTS_Channel, self).__init__()
        if logging:
            self.__glDebugQueue = ProcQueue(10)
        else:
            self.__glDebugQueue = None
        self.__sentPortS = sent_port_s
        self.__sentAddressS = sent_address_s
        self.__recvPortC = recv_port_c
        self.__mainThreadDebug = Debug(self.__glDebugQueue, True, "音频流通道进程(CTS)")
        self.__ErrorStatus = Event()
        self.__nowNums = Value('Q', 0, lock=False)
        self.__nowMs = Value('f', 0.0, lock=False)

    def __taskAudioClientRecv(self, recv_port: int, channel: ThreadQueue):
        self.__mainThreadDebug.pushInfo("[音频流接收器] 客户端音频流接收协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        server.bind(('169.254.171.43', recv_port))
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLIN)
        self.__mainThreadDebug.pushInfo("[音频流接收器] 已进入接收循环")
        lastTime = time.time()
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    time.sleep(0)
                else:
                    recvBytes, _ = server.recvfrom(65535)
                    nowTime = time.time()
                    self.__nowMs.value = round((nowTime - lastTime) * 1000, 2)
                    self.__nowNums.value += 1
                    lastTime = nowTime
                    channel.put(recvBytes)
            except Exception as e:
                self.__mainThreadDebug.pushInfo(
                    f"[音频流接收器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainThreadDebug.pushInfo("[音频流接收器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainThreadDebug.pushInfo("[音频流接收器] 出现全局错误申请, 退出该协程!", "Warning")
        self.__ErrorStatus.set()

    def __taskAudioServerSent(self, sent_port: int, sent_address: str, channel: ThreadQueue):
        self.__mainThreadDebug.pushInfo("[音频流发送器] 服务端音频流发送协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLOUT)
        self.__mainThreadDebug.pushInfo("[音频流发送器] 已进入发送循环...")
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    time.sleep(0)
                else:
                    chunk = channel.get()
                    server.sendto(chunk, (sent_address, sent_port))
            except Exception as e:
                self.__mainThreadDebug.pushInfo(
                    f"[音频流发送器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainThreadDebug.pushInfo("[音频流发送器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainThreadDebug.pushInfo("[音频流发送器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    def run(self) -> None:
        self.__mainThreadDebug.pushInfo("主进程开始运行!", "Debug")
        clientAudioChannel = ThreadQueue(20)

        audioServerSent = Thread(
            target=self.__taskAudioServerSent,
            args=(self.__sentPortS, self.__sentAddressS, clientAudioChannel)
        )
        audioClientRecv = Thread(
            target=self.__taskAudioClientRecv,
            args=(self.__recvPortC, clientAudioChannel)
        )

        audioServerSent.setDaemon(True)
        audioClientRecv.setDaemon(True)
        audioClientRecv.start()
        audioServerSent.start()
        self.__mainThreadDebug.pushInfo("线程运行中!", "Debug")
        audioServerSent.join()
        audioClientRecv.join()
        self.__mainThreadDebug.pushInfo("主进程运行结束!", "Warning")

    def openServer(self):
        self.start()

    def closeServer(self):
        self.__ErrorStatus.set()

    def isOpened(self):
        return self.is_alive()

    def getNums(self):
        return self.__nowNums.value

    def getMs(self):
        return self.__nowMs.value


class SocketChannel(Process):

    def __init__(
            self,
            recv_port_c: int,
            recv_port_s: int,
            sent_port_c: int,
            sent_port_s: int,
            sent_address_s: str = '10.1.0.4',
            sent_address_c: str = '169.254.171.1',
            logging=True
    ):
        super(SocketChannel, self).__init__()
        if logging:
            self.__glDebugQueue = ProcQueue(10)
        else:
            self.__glDebugQueue = None
        self.__recvPortC = recv_port_c
        self.__recvPortS = recv_port_s
        self.__sentPortC = sent_port_c
        self.__sentPOrtS = sent_port_s
        self.__sentAddrS = sent_address_s
        self.__sentAddrC = sent_address_c
        self.__mainDebug = Debug(self.__glDebugQueue, True, "指令|数据通道进程")
        self.__ErrorStatus = Event()

    async def __taskRecvClient(self, recv_port: int, clientQueue: asyncio.Queue):
        self.__mainDebug.pushInfo("[客户端数据接收器] 客户端数据接收协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        server.bind(('169.254.171.43', recv_port))
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLIN)
        self.__mainDebug.pushInfo("[客户端数据接收器] 已进入数据接收循环...")
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    await asyncio.sleep(0)
                else:
                    recvBytes, _ = server.recvfrom(1024)
                    # print(len(recvBytes))
                    await clientQueue.put(recvBytes)
            except Exception as e:
                self.__mainDebug.pushInfo(
                    f"[客户端数据接收器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainDebug.pushInfo("[客户端数据接收器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.pushInfo("[客户端数据接收器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    async def __taskSentClient(self, sent_port: int, sent_address: str, serverQueue: asyncio.Queue):
        self.__mainDebug.pushInfo("[客户端数据发送器] 客户端数据发送协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLOUT)
        self.__mainDebug.pushInfo("[客户端数据发送器] 已进入数据发送循环...")
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    await asyncio.sleep(0)
                else:
                    if serverQueue.empty():
                        await asyncio.sleep(0)
                    else:
                        chunk = await serverQueue.get()
                        server.sendto(chunk, (sent_address, sent_port))
            except Exception as e:
                self.__mainDebug.pushInfo(
                    f"[客户端数据发送器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainDebug.pushInfo("[客户端数据发送器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.pushInfo("[客户端数据发送器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    async def __taskRecvServer(self, recv_port: int, serverQueue: asyncio.Queue):
        self.__mainDebug.pushInfo("[服务端数据接收器] 服务端数据接收协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        server.bind(('10.1.0.5', recv_port))
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLIN)
        self.__mainDebug.pushInfo("[服务端数据接收器] 已进入数据接收循环...")
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    await asyncio.sleep(0)
                else:
                    recvBytes, _ = server.recvfrom(1024)
                    # print(recvBytes)
                    await serverQueue.put(recvBytes)
            except Exception as e:
                self.__mainDebug.pushInfo(
                    f"[服务端数据接收器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainDebug.pushInfo("[服务端数据接收器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.pushInfo("[服务端数据接收器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    async def __taskSentServer(self, sent_port: int, sent_address: str, clientQueue: asyncio.Queue):
        self.__mainDebug.pushInfo("[服务端数据发送器] 服务端数据发送协程初始化...")
        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        myPoll = select.epoll()
        myPoll.register(server, select.EPOLLOUT)
        self.__mainDebug.pushInfo("[服务端数据发送器] 已进入数据发送循环...")
        while not self.__ErrorStatus.is_set():
            try:
                res = myPoll.poll(0)
                if len(res) == 0:
                    await asyncio.sleep(0)
                else:
                    if clientQueue.empty():
                        await asyncio.sleep(0)
                    else:
                        chunk = await clientQueue.get()
                        server.sendto(chunk, (sent_address, sent_port))
            except Exception as e:
                print("I am error")
                self.__mainDebug.pushInfo(
                    f"[服务端数据发送器] \n{traceback.format_exc()}\n----------Error----------",
                    "Error"
                )
                self.__mainDebug.pushInfo("[服务端数据发送器] 该协程发生错误, 退出!", "Error")
                break
        if self.__ErrorStatus.is_set():
            self.__mainDebug.pushInfo("[服务端数据发送器] 出现全局错误申请, 退出该协程!", 'Warning')
        self.__ErrorStatus.set()

    def __mainThread(
            self,
            recv_port_c: int,
            recv_port_s: int,
            sent_port_c: int,
            sent_port_s: int,
            sent_address_s: str = '10.1.0.4',
            sent_address_c: str = '169.254.171.1',
    ):
        loop = asyncio.new_event_loop()
        clientQueue = asyncio.Queue(1, loop=loop)
        serverQueue = asyncio.Queue(1, loop=loop)
        taskRecvClient = loop.create_task(self.__taskRecvClient(recv_port_c, clientQueue))
        taskRecvServer = loop.create_task(self.__taskRecvServer(recv_port_s, serverQueue))
        taskSentClient = loop.create_task(self.__taskSentClient(sent_port_c, sent_address_c, serverQueue))
        taskSentServer = loop.create_task(self.__taskSentServer(sent_port_s, sent_address_s, clientQueue))

        async def main():
            await asyncio.gather(
                taskRecvClient, taskRecvServer, taskSentClient, taskSentServer
            )

        self.__mainDebug.pushInfo("主线程开始运行", "Debug")
        loop.run_until_complete(main())
        self.__ErrorStatus.set()
        self.__mainDebug.pushInfo("退出该线程", "Debug")

    def run(self) -> None:
        mainThread = Thread(
            target=self.__mainThread,
            args=(
                self.__recvPortC, self.__recvPortS, self.__sentPortC, self.__sentPOrtS,
                self.__sentAddrS, self.__sentAddrC
            )
        )
        mainThread.setDaemon(True)
        mainThread.start()
        mainThread.join()

    def openServer(self):
        self.start()

    def closeServer(self):
        self.__ErrorStatus.set()

    def isOpened(self):
        return self.is_alive()


if __name__ == '__main__':
    videoMainRecvServer = Video_StoC_Main(6100, 8103)
    videoSubRecvServer = Video_StoC_Sub(6101, 8104)
    videoMainRecvServer.openServer()
    videoSubRecvServer.openServer()
    while videoMainRecvServer.isOpened() and videoSubRecvServer.isOpened():
        try:
            mainNums, mainMS = videoMainRecvServer.getNums(), videoMainRecvServer.getMs()
            subNums, subMS = videoSubRecvServer.getNums(), videoSubRecvServer.getMs()
            print(f"Main::Nums->{mainNums}|Ms->{mainMS}")
            print(f"Sub::Nums->{subNums}|Ms->{subMS}")
            print("---------")
            time.sleep(1)
        except KeyboardInterrupt:
            break
    videoMainRecvServer.closeServer()
    videoSubRecvServer.closeServer()
    videoMainRecvServer.join()
    videoSubRecvServer.join()

    # audio_stc = Audio_STC_Channel(recv_port_s=6511, sent_port_c=8212)
    # audio_cts = Audio_CTS_Channel(recv_port_c=8211, sent_port_s=6512)
    # audio_stc.start(), audio_cts.start()
    # while audio_stc.isOpened() and audio_cts.isOpened():
    #     try:
    #         print("-----------------------------------------")
    #         stc_nums, stc_ms = audio_stc.getNums(), audio_stc.getMs()
    #         cts_nums, cts_ms = audio_cts.getNums(), audio_cts.getMs()
    #         print(f"stc_nums: {stc_nums} | stc_ms: {stc_ms}")
    #         print(f"cts_nums: {cts_nums} | cts_ms: {cts_ms}")
    #         time.sleep(1)
    #     except KeyboardInterrupt:
    #         break
    # audio_stc.closeServer()
    # audio_cts.closeServer()
    # audio_cts.join()
    # audio_stc.join()
